package com.desheng.bigdata.flink.table

import org.apache.flink.api.scala.{DataSet, ExecutionEnvironment, _}
import org.apache.flink.table.api.Table
import org.apache.flink.table.api.scala.BatchTableEnvironment
import org.apache.flink.types.Row

/**
  * FlinkTable&FlinkSQL核心是TableEnvironment，其核心作用：
  * 1、在内部catalog（元数据管理区）来注册一张表
  * 2、能够注册catalog
  * 3、能够加载可插拔的模块组件
  * 4、执行sql的查询
  * 5、注册用户自定义的函数
  * 6、将dataset和datastream转化为一个table
  * 7、持有executionEnvironment和streamExecutionEnvironment的引用
  *
  *注意： FlinkTable/SQL api中关于TableEnvironment的实现有两种
  *     BatchTableEnvironment 在构建的时候需要持有ExecutionEnvironment
  *     StreamTableEnvironment 在构建的时候需要持有StreamExecutionEnvironment
  *         通过调用create方法来进行实现
  */
object _01FlinkBatchTableOps {
    def main(args: Array[String]): Unit = {
        //创建批对应的Env
        val batchEnv = ExecutionEnvironment.getExecutionEnvironment
        //基于批Env构建BatchTableEnv
        val tblEnv = BatchTableEnvironment.create(batchEnv)

        // load data from collection
        val stus:DataSet[Student] = batchEnv.fromCollection(List(
            "12|张三|25|男|chinese|50",
            "12|张三|25|男|math|60",
            "12|张三|25|男|english|70",
            "12|李四|20|男|chinese|50",
            "12|李四|20|男|math|50",
            "12|李四|20|男|english|50",
            "12|王芳|19|女|chinese|70",
            "12|王芳|19|女|math|70",
            "12|王芳|19|女|english|70",
            "13|张大三|25|男|chinese|60",
            "13|张大三|25|男|math|60",
            "13|张大三|25|男|english|70",
            "13|李大四|20|男|chinese|50",
            "13|李大四|20|男|math|60",
            "13|李大四|20|男|english|50",
            "13|王小芳|19|女|chinese|70",
            "13|王小芳|19|女|math|80",
            "13|王小芳|19|女|english|70"
        )).map(line => {
            val fields = line.split("\\|")
            if(fields == null || fields.length != 6) {
                Student(-1, null, -1, null, null, -1)
            } else {
                val id = fields(0).trim.toInt
                val name = fields(1)
                val age = fields(2).trim.toInt
                val gender = fields(3)
                val course = fields(4)
                val score = fields(5).trim.toDouble
                Student(id, name, age, gender, course, score)
            }
        }).filter(stu => stu.id != -1)
        //create table by dataset
        val stuTbl: Table = tblEnv.fromDataSet(stus)
        //获取table中的schema元数据信息
        stuTbl.printSchema()

        val tbl:Table = stuTbl.select("id, name, age, gender")
        //flink中的table要想执行sink操作，需要将其转化为对应的dataset或者datastream再进行
        tblEnv.toDataSet[Row](tbl).print()

    }
}
//define case class to encapsulate(封装) data
case class Student(id: Int, name: String, age: Int, gender: String, course: String, source: Double)